package xin.marcher.wind.migrate.component.canal;

import cn.hutool.core.collection.CollUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import xin.marcher.framework.common.util.spring.SpringContextUtil;
import xin.marcher.wind.migrate.constants.MigrateConstant;
import xin.marcher.wind.migrate.domain.entity.EtlBinlogConsumeRecordDO;
import xin.marcher.wind.migrate.enums.ConsumerStatus;
import xin.marcher.wind.migrate.mapper.EtlBinlogConsumeRecordMapper;

import java.util.Collection;
import java.util.List;

/**
 * binlog 消息拉取提交任务
 */
@Slf4j
public class CanalPullCommitRunner implements Runnable {

    /**
     * 消息主题
     */
    private final String topic;

    /**
     * rocketmq 的 nameServer 地址
     */
    private final String mqNameServer;

    /**
     * binlog 消息同步消费记录表 Mapper
     */
    private final EtlBinlogConsumeRecordMapper consumeRecordMapper;

    /**
     * 消息拉取提交任务构造方法
     *
     * @param topic         消息主题
     * @param mqNameServer rocketmq 的 nameServer 地址
     */
    public CanalPullCommitRunner(String topic, String mqNameServer) {
        this.topic = topic;
        this.mqNameServer = mqNameServer;
        this.consumeRecordMapper = SpringContextUtil.getBean(EtlBinlogConsumeRecordMapper.class);
    }

    @Override
    public void run() {
        try {
            DefaultLitePullConsumer litePullConsumer = new DefaultLitePullConsumer(MigrateConstant.MQ_BINLOG_COMMIT_GROUP);
            litePullConsumer.setAutoCommit(false);
            litePullConsumer.setNamesrvAddr(mqNameServer);
            litePullConsumer.start();

            commitRun(litePullConsumer);
        } catch (MQClientException e) {
            log.error("消息提交失败", e);
        }
    }

    /**
     * 执行消息提交
     *
     * @param consumer 消息拉取消费者
     */
    private void commitRun(DefaultLitePullConsumer consumer) {
        try {
            Collection<MessageQueue> messageQueues = consumer.fetchMessageQueues(topic);
            consumer.assign(messageQueues);
            try {
                while (true) {
                    // db 中取得所有已消费未提交的记录
                    List<EtlBinlogConsumeRecordDO> consumedRecords = consumeRecordMapper.getNotCommittedConsumedRecords(topic);
                    if (CollUtil.isNotEmpty(consumedRecords)) {
                        for (EtlBinlogConsumeRecordDO consumedRecord : consumedRecords) {
                            // 定位消息, 需要重新消费某个特定消息之前被调用
                            consumer.seek(new MessageQueue(consumedRecord.getTopic(), consumedRecord.getBrokerName(), consumedRecord.getQueueId()), consumedRecord.getOffset());
                            // 这一步必须，不然手动提交的东西不对(做一个 poll 操作, 从指定的位置 poll 拉取过来数据)
                            List<MessageExt> messageExts = consumer.poll();
                            // 提交已消费的消息
                            consumer.commitSync();

                            // 更新消费记录状态为已提交
                            consumedRecord.setConsumeStatus(ConsumerStatus.COMMITTED.getValue());
                            consumeRecordMapper.updateConsumeRecordStatus(consumedRecord);
                        }
                    } else {
                        Thread.sleep(5000);
                    }
                }
            } finally {
                consumer.shutdown();
            }
        } catch (MQClientException | InterruptedException e) {
            try {
                // 异常，这种情况下休眠 5 秒再重试
                Thread.sleep(5000);
                // 执行消息提交
                commitRun(consumer);
            } catch (InterruptedException interruptedException) {
                log.error("消息拉取服务启动失败！", e);
            }
        }
    }
}
